package org.funtester.performance.books.chapter09.section2;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class Consumer {

    public static void main(String[] args) {
        Properties properties = new Properties();// 配置信息
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");// 指定kafka集群地址
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "FunTester");// 指定消费者组
        properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());// key的序列化器
        properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());// value的序列化器
        properties.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");// 自动提交的时间间隔
        properties.setProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "10");// 每次最大拉取的条数
        properties.setProperty(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "10000");// 消费者会话超时时间
        properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");// 消费者组不存在时，从最早的消息开始消费
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);// 创建消费者对象
        String topic = "FunTester";// 指定消费的主题
        consumer.subscribe(Arrays.asList(topic));// 消费者订阅主题
        while (true) {// 消费者拉取数据
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));// 拉取数据,超时时间为1秒,拉取不到数据则返回空
            for (ConsumerRecord<String, String> record : records) {// 遍历拉取到的数据
                System.out.printf("偏移量 %d, key %s, value %s%n", record.offset(), record.key(), record.value());
            }
        }
//        consumer.close();// 关闭消费者对象
    }
}
